﻿// <copyright file="Scheduler.cs" company="Oliver Trenouth">
// Copyright (c) 2013 All Right Reserved, http://olitee.com
//
// All other rights reserved.
//
// THIS CODE AND INFORMATION ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY 
// KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A
// PARTICULAR PURPOSE.
//
// </copyright>
// <author>Oliver Trenouth</author>
// <email>support@cuedeck.com</email>
// <date>2013-03-19</date>
// <summary>An automatic, asynchronous work scheduler for generic items of work.</summary>

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Olitee.WorkScheduler.Interfaces;

namespace Olitee.WorkScheduler
{
    /// <summary>
    /// An automatic, asynchronous work scheduler for generic items of work.
    /// </summary>
    public class Scheduler
    {
        // Internal dictionaries.
        private readonly Dictionary<Type, Type> workProcessors;
        private readonly ConcurrentDictionary<IWorkItem, TaskCompletionSource<IWorkItem>> workItemCompletionSources;

        // Internal work queues
        private readonly ConcurrentQueue<IWorkItem> workInQueue = new ConcurrentQueue<IWorkItem>();
        private readonly ConcurrentBag<IWorkItem> workInProcess = new ConcurrentBag<IWorkItem>();
        private readonly ConcurrentBag<IWorkItem> workCompleted = new ConcurrentBag<IWorkItem>();
        
        // Internal state
        private bool schedulerStarted = false;

        /// <summary>
        /// Creates a generic scheduler.
        /// </summary>
        /// <param name="maximumConcurrentProcesses">The maximum number of concurrent processes. If set to -1, the
        /// value will be automatically set to the number of logical CPU cores.</param>
        public Scheduler()
        {
            workProcessors = new Dictionary<Type, Type>();
            workItemCompletionSources = new ConcurrentDictionary<IWorkItem, TaskCompletionSource<IWorkItem>>();
            this.SchedulerPollInterval = TimeSpan.FromSeconds(1);
            this.MaximumConcurrentWorkProcesses = Environment.ProcessorCount;
        }

        /// <summary>
        /// Gets or sets the maximum number of work items that can be processed concurrently.
        /// Defaults to the number of available logical processors. This can be increased at any time.
        /// Decreasing while the scheduler is running will not affect any currently running work items.
        /// </summary>
        public int MaximumConcurrentWorkProcesses { get; set; }

        /// <summary>
        /// Gets of sets the time the scheduler to pause before checking the queue for new/more work. Default: 1000ms.
        /// </summary>
        public TimeSpan SchedulerPollInterval { get; set; }

        /// <summary>
        /// Registers and associates a concrete IWorkProcessor with a concrete IWorkItem
        /// </summary>
        /// <typeparam name="TWorkItem"></typeparam>
        /// <typeparam name="TWorkProcessor"></typeparam>
        public void RegisterWorkProcessor<TWorkItem, TWorkProcessor>()
            where TWorkItem : class, IWorkItem
            where TWorkProcessor : class, IWorkProcessor, new()
        {
            if (!workProcessors.ContainsKey(typeof(TWorkItem)))
            {
                workProcessors.Add(typeof(TWorkItem), typeof(TWorkProcessor));
            }
        }

        /// <summary>
        /// Starts the scheduler service on a dedicated thread, and begins monitoring the work queue.
        /// </summary>
        public void StartScheduler()
        {
            if (!schedulerStarted)
            {
                schedulerStarted = true;

                // Create a dedicated thread for the scheduler
                Task.Factory.StartNew(() =>
                {
                    // Main scheduler loop
                    while (this.schedulerStarted)
                    {
                        // Limit the parallellism
                        if (this.workInProcess.Count <= this.MaximumConcurrentWorkProcesses)
                        {
                            DequeueAndProcessNextWorkItem();
                        }

                        // Pause the queue monitoring for a short time
                        System.Threading.Thread.Sleep(this.SchedulerPollInterval);
                    }
                }, TaskCreationOptions.LongRunning);
            }
        }

        /// <summary>
        /// Stops the scheduler service. Any work items currently being processed will continue until finished,
        /// but no further work items will be processed.
        /// </summary>
        public void StopScheduler()
        {
            if (schedulerStarted)
            {
                schedulerStarted = false;
            }
        }

        /// <summary>
        /// Returns a single list of all work items currently managed by the scheduler.
        /// </summary>
        public List<IWorkItem> AllWorkItems
        {
            get
            {
                // Compile the separate IWorkItem collections into a single list
                var allWorkItemsList = new List<IWorkItem>();
                workInQueue.ToList().ForEach(w => allWorkItemsList.Add(w));
                workInProcess.ToList().ForEach(w => allWorkItemsList.Add(w));
                workCompleted.ToList().ForEach(w => allWorkItemsList.Add(w));
                return allWorkItemsList;
            }
        }

        public Task<IWorkItem> StartWorkItemImmediately(IWorkItem workItem)
        {
            workCompleted.Add(workItem);
            IWorkProcessor processor = GetWorkProcessor(workItem);
            return processor.Start(workItem);
        }

        /// <summary>
        /// Adds a work item to the queue.
        /// </summary>
        /// <param name="workItem"></param>
        /// <returns>A task representing the work item.</returns>
        public Task<IWorkItem> AddWorkItem(IWorkItem workItem)
        {
            this.workInQueue.Enqueue(workItem);

            var tcs = new TaskCompletionSource<IWorkItem>();

            // Add the TCS to the completeion source dictionary.
            // When this work is dequeued by the scheduler, it will fetch this TCS and set it's result
            // once the work is comeplete.
            this.workItemCompletionSources.TryAdd(workItem, tcs);

            // Return the task so the calling code can hook up continuations, waits, etc.
            return tcs.Task;
        }

        /// <summary>
        /// Returns an instance of a work processor.
        /// </summary>
        /// <param name="workItem">The IWorkItem that a suitable IWorkProcessor should be returned for.</param>
        /// <returns></returns>
        protected virtual IWorkProcessor GetWorkProcessor(IWorkItem workItem)
        {
            Type workItemType = workItem.GetType();
            if (workProcessors.ContainsKey(workItemType))
            {
                Type workProcessorType = workProcessors[workItem.GetType()];
                IWorkProcessor newProcessor = Activator.CreateInstance(workProcessorType) as IWorkProcessor;
                return newProcessor;
            }
            else
            {
               throw new Exception(String.Format("An IWorkProcessor for a work item of type {0} was not found.", workItem.GetType()));
            }
        }

        /// <summary>
        /// Obtains the next item in the queue, obtains a work processor, and begins the work.
        /// </summary>
        private Task<IWorkItem> DequeueAndProcessNextWorkItem()
        {
            IWorkItem currentWorkItem;
            this.workInQueue.TryDequeue(out currentWorkItem);

            if (currentWorkItem != null)
            {
                // Move the currentWorkItem into the InProgress list
                this.workInProcess.Add(currentWorkItem);

                Task<IWorkItem> workTask;

                try
                {
                    var work = GetWorkProcessor(currentWorkItem);
                    workTask = work.Start(currentWorkItem);

                }
                catch (Exception ex)
                {
                    currentWorkItem.Status = Enumerations.WorkItemStatus.Error;
                    currentWorkItem.Message = ex.Message;
                    workTask = new Task<IWorkItem>(new Func<IWorkItem>(() => { return null; }));
                }
                    
                workTask.ContinueWith(t =>
                {
                    // Once complete, fetch the TCS created when the job was attached, and set it's result.
                    TaskCompletionSource<IWorkItem> tcs;
                    this.workItemCompletionSources.TryRemove(currentWorkItem, out tcs);

                    // Take the item from the InProcess list
                    IWorkItem finishedWork;
                    this.workInProcess.TryTake(out finishedWork);

                    if (finishedWork != null)
                    {
                        // Add the item to the completed work list.
                        this.workCompleted.Add(finishedWork);
                    }

                    if (tcs != null)
                    {
                        // Set the result of the task that was returned
                        // when the AddWorkItem() method was called.
                        tcs.SetResult(currentWorkItem);
                    }
                });

                return workTask;
            }

            return null;
        }
    }
}